package beautiful.butterfly.drds.data_exchange.transport.exchanger;


import beautiful.butterfly.drds.data_exchange.constant.Constants;
import beautiful.butterfly.drds.data_exchange.exception.CommonErrorCode;
import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.plugin.TaskPluginCollector;
import beautiful.butterfly.drds.data_exchange.transport.record.DefaultRecord;
import beautiful.butterfly.drds.data_exchange.transport.record.TerminateRecord;
import beautiful.butterfly.drds.data_exchange.transport.record.column.Record;
import beautiful.butterfly.drds.data_exchange.transport.record_queue.RecordQueue;
import beautiful.butterfly.drds.data_exchange.util.Configuration;
import org.apache.commons.lang3.Validate;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;


public class QueueExchanger implements Produce, Consume
{


    protected final int byteCapacity;
    //
    private final RecordQueue recordQueue;
    private final List<Record> recordList;
    private final Configuration configuration;
    private final AtomicInteger memoryByteSize = new AtomicInteger(0);
    //
    private final TaskPluginCollector taskPluginCollector;
    private int recordListSize;
    private int recordListIndex = 0;
    private volatile boolean shutdown = false;

    @SuppressWarnings("unchecked")
    public QueueExchanger(final RecordQueue recordQueue, final TaskPluginCollector taskPluginCollector)
    {
        assert null != recordQueue;
        assert null != recordQueue.getConfiguration();

        this.recordQueue = recordQueue;
        this.taskPluginCollector = taskPluginCollector;
        this.configuration = recordQueue.getConfiguration();

        this.recordListSize = configuration
                .getInt(Constants.transport_exchanger_buffersize);
        this.recordList = new ArrayList<Record>(recordListSize);

        //channel的queue默认大小为8M，原来为64M
        this.byteCapacity = configuration.getInt(Constants.transport_channel_capacity_byte, 8 * 1024 * 1024);


    }

    /**
     * <pre>
     * ========================================================================================================================
     * Produce
     * ========================================================================================================================
     * </pre>
     */

    public Record createRecord()
    {
        return new DefaultRecord();
    }


    public void addToRecordList(Record record)
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }

        Validate.notNull(record, "record不能为空.");

        if (record.getMemoryByteSize() > this.byteCapacity)
        {
            this.taskPluginCollector.collectDirtyRecord(record, new Exception(String.format("单条记录超过大小限制，当前限制为:%s", this.byteCapacity)));
            return;
        }

        boolean isFull = (this.recordListIndex >= this.recordListSize || this.memoryByteSize.get() + record.getMemoryByteSize() > this.byteCapacity);
        if (isFull)
        {
            pushAllToRecordQueue();
        }

        this.recordList.add(record);
        this.recordListIndex++;
        memoryByteSize.addAndGet(record.getMemoryByteSize());
    }


    public void pushAllToRecordQueue()
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }
        this.recordQueue.pushAll(this.recordList);
        this.recordList.clear();
        this.recordListIndex = 0;
        this.memoryByteSize.set(0);
    }


    public void terminate()
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }
        pushAllToRecordQueue();
        this.recordQueue.pushTerminateRecord(TerminateRecord.get());
    }

    public void shutdown()
    {
        shutdown = true;
        try
        {
            recordList.clear();
            recordQueue.clear();
        } catch (Throwable t)
        {
            t.printStackTrace();
        }
    }

    /**
     * <pre>
     * ========================================================================================================================
     * Consume
     * ========================================================================================================================
     * </pre>
     */

    public Record getFromRecordList()
    {
        if (shutdown)
        {
            throw DataExchangeException.asDataExchangeException(CommonErrorCode.SHUT_DOWN_TASK, "");
        }
        boolean isEmpty = (this.recordListIndex >= this.recordList.size());
        if (isEmpty)
        {
            pullAllFromRecordQueue();
        }

        Record record = this.recordList.get(this.recordListIndex++);
        if (record instanceof TerminateRecord)
        {
            record = null;
        }
        return record;
    }


    private void pullAllFromRecordQueue()
    {
        this.recordQueue.pullAll(this.recordList);
        this.recordListIndex = 0;
        this.recordListSize = this.recordList.size();
    }
}
